package com.shengzai.mapreduce.groupby;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.stream.Collector;


public class ReduceGroupBy extends Reducer<Text,Text,Text,Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        ArrayList<Student> list = new ArrayList<>();
        int tmp=0;
        int count=0;
        for (Text value : values) {
            String s = value.toString();
            String[] split = s.split(",");
            Student student = new Student(split[0],split[1],split[2],Integer.parseInt(split[3]));
            list.add(student);
        }
        Collections.sort(list, new Comparator<Student>() {
            @Override
            public int compare(Student o1, Student o2) {
                return -((o1.getScore() < o2.getScore()) ? -1 : ((o1.getScore() == o2.getScore()) ? 0 : 1));
            }
        });
        for (Student student : list) {
            if (count<list.size()-1) {
                tmp = student.getScore() - list.get(count + 1).score;
                count++;
                context.write(key, new Text(student.getsID() + "\t" + student.getName() + "\t" + student.getScore() + "\t" + tmp));
            }else {
                context.write(key, new Text(student.getsID() + "\t" + student.getName() + "\t" + student.getScore()));
            }

        }

    }
}
